import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api .*;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions .*;


public class Overview_Examples
{

    public static void main(String[] args) throws Exception
    {

// environment configuration
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);


        DataSet<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4)
               ));


// register Orders table in table environment

        Table Orders = tEnv.fromDataSet(orderA,"user,product,amount");
//        tEnv.registerTable("Orders",tableA);


//        Table orders=tEnv.from("Or");

// specify table program
//    Table orders = tEnv.from("Orders"); // schema (a, b, c, rowtime)

        Table counts = Orders
              .groupBy($("user"),$("product"))
                .select($("user"),$("product"),$("amount").sum().as("cnt"));

// conversion to DataSet
    DataSet<Row> result = tEnv.toDataSet(counts, Row.class);
    result.print();

    }
}
